package com.gugee.ins.data.collector.bloggerfollowing.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.constant.InsBloggerMsgTypeCodeEnum;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.blogger.InsRelBloggerFollowingDao;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsRelBloggerFollowing;
import com.gugee.ins.common.model.spider.blogger.EdgeBlogger;
import com.gugee.ins.common.model.spider.blogger.InsBloggerFollowingResponse;
import com.gugee.ins.common.model.spider.blogger.PageInfo;
import com.gugee.ins.data.collector.bloggerfollowing.config.InsBloggerFollowingConfig;
import com.gugee.ins.data.collector.bloggerfollowing.service.InsBloggerFollowingService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import lombok.val;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerFollowingServiceImpl implements InsBloggerFollowingService {

    /**
     * 配置
     */
    @Autowired
    InsBloggerFollowingConfig insBloggerFollowingConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * insBloggerMainDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     * InsRelBloggerActiveFollowerDao
     */
    @Autowired
    InsRelBloggerFollowingDao insRelBloggerFollowingDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insBloggerFollowing-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insBloggerFollowingConfig.getInsBloggerFollowingCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理 活跃粉丝的关注列表 采集 任务
     * @param insBloggerMessages
     */
    @Override
    public void handleActiveFollowerFollowingCollect(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runCollectActiveFollowerFollowing(msg);
                } catch (Exception ex) {
                    log.error("Active Follower Following collect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("Active Follower Following collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集 活跃粉丝的关注列表
     * @param insBloggerMessage
     */
    public void runCollectActiveFollowerFollowing(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();
        // 时间范围
        long limitTime = start / 1000 - insBloggerFollowingConfig.getInsBloggerFollowingClIntervalDay() * 24 * 3600;
        // 判断类型，是否需要查询大播主自己的关注列表
        if(insBloggerMessage.getTypeCode() != InsBloggerMsgTypeCodeEnum.NO_HANDLE_AND_DO_OTHER_OPT.getCode()){
            try{
                runCollectActiveFollowerFollowing(uid);
            }catch (Exception e){
                log.error(" insMainBlogger Following collect error. uid:{}, error :{}",uid,e.getMessage());
            }
        }

        // 查询他的活跃粉丝，然后发送爬取关注列表任务消息
        List<InsBloggerMessage> subMessages = insBloggerMainDao.findInsBloggerFollowingCollectMessageByUid(insBloggerMessage.getUid(),
                limitTime,insBloggerFollowingConfig.getInsBloggerFollowingClMinCount(),insBloggerFollowingConfig.getInsBloggerFollowingClActiveFollowerCount());

        if(subMessages != null && !subMessages.isEmpty()){
            try {
                forkJoinPool.submit(() -> subMessages.parallelStream().forEach(msg -> {
                    try {
                        runCollectActiveFollowerFollowing(msg.getUid());
                    } catch (Exception ex) {
                        log.error("subBlogger Active Follower Following collect error. uid:{}, error :{}" ,
                                msg.getUid(),ex.getMessage());
                    }
                })).get();
            } catch (Exception ex) {
                log.error("subBlogger Active Follower Following collect  thread error.  {}" ,  ex.getMessage());
            }

            // 发送优先统计品牌的消息
            sendPriorityStMsg(insBloggerMessage);
        }

        if ((System.currentTimeMillis() - start) / 1000 > insBloggerFollowingConfig.getBloggerFollowingDebugTimeLimit()) {
            log.info("Active Follower Following Collect uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 采集 活跃粉丝的关注列表
     * @param uid
     */
    public void runCollectActiveFollowerFollowing(long uid){
        // 根据uid爬取信息
        PageInfo pageInfo;
        Set<Long> insRelBloggerFollowingUid = new HashSet<>(100);
        List<InsRelBloggerFollowing> insRelBloggerFollowings = new LinkedList<>();

        pageInfo = spiderFollowing(uid,insBloggerFollowingConfig.getInsSpiderBloggerFollowingEachCount(),"",insRelBloggerFollowingUid,insRelBloggerFollowings);
        while(pageInfo != null && pageInfo.getHasNextPage()
                && !StringUtils.isEmpty(pageInfo.getEndCursor())
                && insRelBloggerFollowingUid.size() < insBloggerFollowingConfig.getInsActiveFollowerFollowingCount()){
            pageInfo = spiderFollowing(uid,insBloggerFollowingConfig.getInsSpiderBloggerFollowingEachCount(),pageInfo.getEndCursor(),insRelBloggerFollowingUid,insRelBloggerFollowings);
        }


        if(!insRelBloggerFollowingUid.isEmpty()){
            // 更新博主与关注人的关系
            try{
                save(uid,insRelBloggerFollowings);
            }catch (Exception e){
                log.error("update Active Follower Following error.msg : {}" ,e.getMessage());
            }
            // 更新标记位
            insBloggerDao.updateFollowingCollectTime(uid);
        }
    }

    /**
     * 循环拉关注人
     * @param uid
     * @param count
     * @param endCursor
     * @param insRelBloggerFollowingUid
     * @return
     */
    private PageInfo spiderFollowing(long uid,int count,String endCursor,Set<Long> insRelBloggerFollowingUid,List<InsRelBloggerFollowing> insRelBloggerFollowings){
        //根据uid爬取信息
        List<EdgeBlogger> edges;
        InsBloggerFollowingResponse insBloggerFollowingResponse;
        PageInfo pageInfo;
        long nowTime = System.currentTimeMillis() / 1000;
        try {
            insBloggerFollowingResponse = insBloggerSpiderClient.getFollowingBloggerByUid(uid,count,endCursor);
            edges = insBloggerFollowingResponse.getData().getUser().getEdgeFollow().getEdges();
            if(edges != null && !edges.isEmpty()){
                for(EdgeBlogger edge : edges){
                    if(edge != null && edge.getNode() != null
                            && !edge.getNode().getIsPrivate()
                            && edge.getNode().getUid() != null
                            && edge.getNode().getUid() > 0
                            && !insRelBloggerFollowingUid.contains(edge.getNode().getUid())){
                        insRelBloggerFollowingUid.add(edge.getNode().getUid());
                        InsRelBloggerFollowing insRelBloggerFollowing = new InsRelBloggerFollowing();
                        val insBlogger = insBloggerDao.findByUid(edge.getNode().getUid());
                        if(insBlogger != null){
                            // 设置是否是品牌
                            if(insBlogger.getIsBusiness()){
                                insRelBloggerFollowing.setFType(1);
                            }else if(StringUtils.isEmpty(insBlogger.getFullName())){
                                insRelBloggerFollowing.setFType(3);
                            }else{
                                insRelBloggerFollowing.setFType(2);
                            }
                        }else{
                            sendPriorityCollectBloggerMsg(edge.getNode().getUid(),edge.getNode().getUsername(),edge.getNode().getPicUrl());
                        }
                        insRelBloggerFollowing.setUid(uid);
                        insRelBloggerFollowing.setFUid(edge.getNode().getUid());
                        insRelBloggerFollowing.setCreateTime(nowTime);
                        insRelBloggerFollowings.add(insRelBloggerFollowing);
                        if(insRelBloggerFollowingUid.size() >= insBloggerFollowingConfig.getInsActiveFollowerFollowingCount()){
                            break;
                        }
                    }
                }
            }
            pageInfo = insBloggerFollowingResponse.getData().getUser().getEdgeFollow().getPageInfo();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider Active Follower Following error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     * 发送博主优先收录的消息
     * @param uid
     * @param userName
     * @param picUrl
     */
    public void sendPriorityCollectBloggerMsg(long uid,String userName,String picUrl){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,false)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRetryLimitTimes(insBloggerFollowingConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insBloggerFollowingConfig.getInsBloggerPriorityRetryTimes());

            kafkaTemplateInsBlogger.send(insBloggerFollowingConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);

        }
    }

    /**
     * 发送优先统计品牌的消息
     * @param insBloggerMessage
     */
    public void sendPriorityStMsg(InsBloggerMessage insBloggerMessage){
        insBloggerMessage.setTypeCode(InsBloggerMsgTypeCodeEnum.ST_BRAND.getCode());
        insBloggerMessage.setRetryTimes(insBloggerFollowingConfig.getInsMBloggerActiveFollowerDataStPriorityRetryTimes());
        insBloggerMessage.setRetryLimitTimes(insBloggerFollowingConfig.getInsMBloggerActiveFollowerDataStPriorityRetryTimes());
        insBloggerMessage.setEventTime(System.currentTimeMillis());
        kafkaTemplateInsBlogger.send(insBloggerFollowingConfig.getInsMBloggerActiveFollowerDataStPriorityTopic(),insBloggerMessage);
    }

    /**
     * 删除原来数据并保存当前数据
     * @param uid
     * @param insRelBloggerFollowings
     */
    @Transactional(rollbackFor = Exception.class)
    public void save(long uid,List<InsRelBloggerFollowing> insRelBloggerFollowings){
        insRelBloggerFollowingDao.deleteByUid(uid);
        insRelBloggerFollowingDao.batchInsertByUid(insRelBloggerFollowings);
    }
}
